package com.swak.paxos.impl;

import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.springframework.util.Assert;

import com.swak.paxos.Instance;
import com.swak.paxos.PaxosConfig;
import com.swak.paxos.Value;
import com.swak.paxos.breakpoint.InstanceBP;
import com.swak.paxos.enums.ProposalType;
import com.swak.paxos.protocol.PaxosForward;
import com.swak.paxos.protocol.PaxosMessage;
import com.swak.paxos.statemachine.StateMachineFac;
import com.swak.paxos.statemachine.master.MasterStateMachine;
import com.swak.paxos.statemachine.system.SystemxStateMachine;
import com.swak.paxos.store.PaxosStorage;
import com.swak.reactivex.threads.EventLoop;
import com.swak.reactivex.transport.Message;
import com.swak.timer.RTO;
import com.swak.utils.Longs;
import com.swak.utils.Queues;

import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;

/**
 * Instance.
 * 
 * @author DELL
 */
@Getter
public class PaxosInstance implements Instance {

	private final PaxosNode node;
	private final PaxosGroup group;
	private final PaxosConfig config;
	private final PaxosProposer proposer;
	private final PaxosAcceptor acceptor;
	private final PaxosLearner learner;
	private final EventLoop eventLoop;
	private final PaxosStorage storage;
	private final SystemxStateMachine systemSM;
	private final MasterStateMachine masterSM;
	private final StateMachineFac smFAC;

	private final Queue<CommitContext> committingQueue;
	private final AtomicBoolean committing = new AtomicBoolean(false);

	private long commitTimeoutMs;
	private final RTO timeoutRTO = new RTO();

	/**
	 * 当前实例的ID
	 */
	private long instanceID;

	public PaxosInstance(PaxosConfig config, PaxosNode node, PaxosGroup group, PaxosStorage storage,
			EventLoop eventLoop, SystemxStateMachine systemSM, MasterStateMachine masterSM) {
		this.node = node;
		this.group = group;
		this.config = config;
		this.storage = storage;
		this.eventLoop = eventLoop;
		this.systemSM = systemSM;
		this.masterSM = masterSM;
		this.smFAC = new StateMachineFac(config.getSmFac().getSms(group.getGroupID())).addStateMachine(systemSM)
				.addStateMachine(masterSM);
		this.committingQueue = Queues.newMpscQueue();
		this.commitTimeoutMs = this.config.getCommitTimeoutMs();
		this.proposer = new PaxosProposer(config, node, group, this, eventLoop, storage);
		this.acceptor = new PaxosAcceptor(config, node, group, this, storage);
		this.learner = new PaxosLearner(config, node, group, this, this.acceptor, eventLoop, storage);
	}

	/**
	 * 提交一个值
	 * 
	 * @param value
	 */
	@Override
	public CompletableFuture<Void> propose(Value value) {

		/*
		 * 创建一个提交任务
		 */
		CommitContext commitContext = new CommitContext(value, this.commitTimeoutMs, this.config.getTimeUnit());
		this.committingQueue.add(commitContext);

		/*
		 * 执行提交任务
		 */
		if (this.committing.compareAndExchange(false, true) == false) {
			doInnerPropose();
		}

		/*
		 * 返回异步结果, 提交者可以等待提交结果, 或者取消提交任务
		 */
		return commitContext.getCommitFuture();
	}

	void doInnerPropose() {

		/*
		 * 必须保证 committing 处于正确的状态
		 */
		Assert.isTrue(committing.get(), "Committing state not corrent.");

		/*
		 * 执行任务，且必须处于 eventLoop 中执行.
		 */
		if (this.eventLoop.inEventLoop()) {
			for (;;) {

				// 检查 Commit Context
				CommitContext commitContext = this.committingQueue.peek();

				// 没有任务，则退出发送循环
				if (commitContext == null) {
					committing.set(false);
					return;
				}

				// 任务被取消，则删除此任务， 继续下一个任务
				if (commitContext.isEnd()) {
					this.committingQueue.poll();
					continue;
				}

				// 提交任务
				try {
					this.doCommit(commitContext);
					return;
				} catch (Exception e) {

					// 删除此任务
					this.committingQueue.poll();

					// 设置任务失败
					commitContext.error(e);
				}
			}
		} else {
			this.eventLoop.execute(() -> {
				doInnerPropose();
			});
		}
	}

	/**
	 * 提交值：
	 * 
	 * 提交之前需要判断系统状态， 只有系统状态一致才能提交
	 * 
	 * 
	 * @param commitContext
	 */
	void doCommit(CommitContext commitContext) {

		/*
		 * 如果消息是不需要发送的，则直接返回， 其他的节点会提交此消息
		 */
		if (commitContext.getValue().isNoSend()) {
			return;
		}

		/*
		 * 系统状态如果不一致，则需要先同步系统状态
		 */
		if (!this.systemSM.isAccordant()) {
			this.proposer.commit(Value.of().setNodeID(this.node.id()).setSmID(this.systemSM.getSmID())
					.setValue(this.systemSM.needAccordantValue()));
			return;
		}

		/*
		 * 提交数据
		 */
		this.proposer.commit(commitContext.getValue());
	}

	/**
	 * 收到消息
	 * 
	 * @param message
	 */
	@Override
	public void onReceiveMessage(Message message) {

		Object data = message.getData();

		/*
		 * 处理 paxos 消息
		 */
		if (data instanceof PaxosMessage) {
			this.doInnerReceivePaxosMessage((PaxosMessage) data);
		}

		/*
		 * 处理 转发 消息
		 */
		else if (data instanceof PaxosForward) {
			this.propose(((PaxosForward) data).getValue());
		}
	}

	/**
	 * 处理 PaxosMessage 消息
	 * 
	 * @param message
	 */
	void doInnerReceivePaxosMessage(PaxosMessage message) {

		InstanceBP.onReceivePaxosMsg(this.group.getGroupID(), this.instanceID, message);

		/*
		 * Proposer.
		 */
		if (ProposalType.forProposer(message.getType())) {
			if (!this.systemSM.inMemberShip(message.getNodeID())) {
				return;
			}
			this.doInnerReceivePaxosMessageForProposer(message);
		}

		/*
		 * Acceptor.
		 */
		else if (ProposalType.forAcceptor(message.getType())) {
			if (!this.systemSM.inMemberShip(message.getNodeID())) {
				return;
			}
			this.doInnerReceivePaxosMessageForAcceptor(message);
		}

		/*
		 * Learner.
		 */
		else if (ProposalType.forLearner(message.getType())) {
			this.doInnerReceivePaxosMessageForLearner(message);
		}

		/*
		 * Error Type.
		 */
		else {
			InstanceBP.onReceivePaxosMsgTypeNotValid(this.group.getGroupID(), this.instanceID);
		}
	}

	void doInnerReceivePaxosMessageForProposer(PaxosMessage message) {

		/*
		 * follower 节点不参与 Paxos 协议各阶段.
		 * follower 节点只接受数据对齐
		 */
		if (this.group.isIMFollower()) {
			return;
		}

		/*
		 * 消息中的 InstanceID 和 当前的 InstanceID 值不一致.
		 */
		if (message.getInstanceID() != this.getInstanceID()) {

			// TODO 这里有一步记录过期的投票步骤，暂时不知道啥意思.
			InstanceBP.onReceivePaxosAcceptorMsgInotsame(this.group.getGroupID(), this.instanceID);
			return;
		}

		/*
		 * 收到 第一阶段的回复的消息
		 */
		if (message.getType() == ProposalType.PrepareReply) {
			this.proposer.onPrepareReply(message);
		}

		/*
		 * 收到 第二阶段的回复的消息
		 */
		else if (message.getType() == ProposalType.AcceptReply) {
			this.proposer.onAcceptReply(message);
		}
	}

	void doInnerReceivePaxosMessageForAcceptor(PaxosMessage message) {

		/*
		 * follower 节点不参与 Paxos 协议各阶段.
		 */
		if (this.group.isIMFollower()) {
			return;
		}

		/*
		 * 校验 InstanceID
		 */
		if (message.getInstanceID() != this.getInstanceID()) {
			InstanceBP.onReceivePaxosProposerMsgInotsame(this.group.getGroupID(), this.instanceID);
			return;
		}

		// TODO 这里有一步记录过期的投票步骤，暂时不知道啥意思.

		/*
		 * 第一阶段
		 */
		if (message.getType() == ProposalType.Prepare) {
			this.acceptor.onPrepare(message);
		}

		/*
		 * 第二阶段
		 */
		else if (message.getType() == ProposalType.Accept) {
			this.acceptor.onAccept(message);
		}
	}

	void doInnerReceivePaxosMessageForLearner(PaxosMessage message) {

		/*
		 * 本 Node 学习一个值.
		 */
		if (message.getType() == ProposalType.Proposed) {
			this.learner.onProposed(message);
		}

		/*
		 * 这个消息是 Learner 学习结束后 发送的 主要是发送给 Follower
		 */
		else if (message.getType() == ProposalType.SendLearnValue
				|| message.getType() == ProposalType.SendLearnValueNeedAck) {
			this.learner.onSendLearnValue(message);
		}

		/*
		 * 这个消息是 Learner 学习结束后 发送的 主要是发送给 Follower -- Ack
		 */
		else if (message.getType() == ProposalType.SendLearnValueAck) {
			this.learner.onSendLearnValueAck(message);
		}

		/*
		 * 主动发起学习
		 */
		else if (message.getType() == ProposalType.AskforLearn) {
			this.learner.onAskforLearn(message);
		}
		
		/*
		 * 主动学习到的实例ID
		 */
		else if (message.getType() == ProposalType.SendNowInstanceID) {
			this.learner.onSendNowInstanceID(message);
		}

		/*
		 * 如果已近学习到最新的值，则可以开启新的一轮
		 */
		if (!this.learner.isLearned()) {
			return;
		}

		/*
		 * 学习到的值
		 */
		Value value = this.learner.getLearnedValue();

		/*
		 * 需要： 状态机执行成功
		 */
		if (!this.smFAC.execute(this.group.getGroupID(), instanceID, value)) {
			return;
		}

		/*
		 * 已经学习到最新的值
		 */
		InstanceBP.onInstanceLearned(this.group.getGroupID(), this.instanceID);

		/*
		 * 开启新的一轮
		 */
		this.newInstance();

		/*
		 * 开启新的一轮
		 */
		InstanceBP.newInstance(this.group.getGroupID(), this.instanceID);

		/*
		 * 继续提交新的数据，否则超时处理.
		 */
		this.continueCommit(value);
	}

	private void newInstance() {
		this.instanceID++;
		this.proposer.newInstance();
		this.acceptor.newInstance();
		this.learner.newInstance();
	}

	private void continueCommit(Value ownValue) {
		try {

			/*
			 * 执行
			 */
			CommitContext commitContext = this.committingQueue.peek();
			if (commitContext == null) {
				return;
			}

			/*
			 * 是否提交成功
			 */
			boolean success = false;

			/*
			 * 请求ID、 NodeID 一致； 不需要校验 NodeID 和本机是否一致（可以通过Master 来提交消息）
			 */
			if (commitContext.getValue().getId() == ownValue.getId()
					&& commitContext.getValue().getNodeID().equals(ownValue.getNodeID())) {
				this.committingQueue.poll();
				success = true;
			}

			/*
			 * 如果提交成功，则给予反馈，否则继续处理
			 */
			if (success) {
				commitContext.success();
			}
		} finally {
			/*
			 * 继续后续的处理
			 */
			this.doInnerPropose();
		}

	}

	private void calculateTimeoutRTO(long useTimeMs) {
		this.timeoutRTO.calculate(useTimeMs);
		this.commitTimeoutMs = timeoutRTO.getTimeoutInerval();
		this.commitTimeoutMs = Longs.bound(this.config.getCommitTimeoutMs(), this.commitTimeoutMs,
				this.config.getMaxCommitTimeoutMs());
	}

	private void cancelCommitContext(CommitContext context) {
		if (!this.eventLoop.inEventLoop()) {
			this.eventLoop.execute(() -> {
				cancelCommitContext(context);
			});
			return;
		}

		// 取消
		context.cancel();
	}

	/**
	 * CommitContext.
	 * 
	 * @author 618lf
	 */
	@Getter
	@Setter
	@Accessors(chain = true)
	class CommitContext {
		private final Value value;
		private final Runnable timeoutRunnable;
		private final long startMS = System.currentTimeMillis();
		private final CompletableFuture<Void> commitFuture;
		private final ScheduledFuture<?> timeoutFuture;
		private final CompletableFuture<Void> cancelFuture;
		private final long commitTimeoutMs;

		CommitContext(Value value, long commitTimeoutMs, TimeUnit timeUnit) {
			this.commitFuture = new CompletableFuture<Void>();
			this.value = value;
			this.commitTimeoutMs = commitTimeoutMs;
			this.timeoutRunnable = this::onTimeout;
			this.timeoutFuture = PaxosInstance.this.eventLoop.schedule(timeoutRunnable, this.commitTimeoutMs, timeUnit);

			// 监听取消
			this.cancelFuture = this.commitFuture.whenComplete((r, e) -> {

				// 一般是外部取消
				if (this.commitFuture.isCancelled()) {
					cancelCommitContext(this);
				}
			});
		}

		void onTimeout() {
			InstanceBP.commitTimeout(this.commitTimeoutMs, PaxosInstance.this.group.getGroupID(),
					PaxosInstance.this.instanceID);
			this.error(new TimeoutException("commit value timeout!"));
		}

		void cancel() {
			this.success();
		}

		boolean isEnd() {
			return this.timeoutFuture.isDone() || timeoutFuture.isDone();
		}

		void end() {
			long useTimeMs = System.currentTimeMillis() - startMS;
			calculateTimeoutRTO(useTimeMs);
		}

		void success() {
			try {
				if (!timeoutFuture.isCancelled() && !timeoutFuture.isDone()) {
					timeoutFuture.cancel(false);
				}
			} catch (Exception e) {
				// ignore
			}
			try {
				if (!commitFuture.isCancelled() && !commitFuture.isDone()) {
					commitFuture.complete(null);
				}
			} catch (Exception e) {
				// ignore
			}
			this.end();
		}

		void error(Throwable ex) {
			try {
				if (!timeoutFuture.isCancelled() && !timeoutFuture.isDone()) {
					timeoutFuture.cancel(false);
				}
			} catch (Exception e) {
				// ignore
			}
			try {
				if (!commitFuture.isCancelled() && !commitFuture.isDone() && !commitFuture.isCompletedExceptionally()) {
					commitFuture.completeExceptionally(ex);
				}
			} catch (Exception e) {
				// ignore
			}
			this.end();
		}
	}
}
